package com.gy.spark.sparkstreaming;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

public class StreamingWordCount {

    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setAppName(StreamingWordCount.class.getSimpleName()).setMaster("local[*]");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(10));

        JavaReceiverInputDStream<String> dStream = jsc.socketTextStream("localhost", 9999);


        JavaPairDStream<String, Integer> resultDStream = dStream.flatMap(x -> Arrays.asList(x.split(",")).iterator())
                .mapToPair(x -> new Tuple2<String, Integer>(x, 1))
                .reduceByKey(Integer::sum);

        resultDStream.print();

        jsc.start();
        jsc.awaitTermination();
        jsc.stop();
    }
}
